package com.qs.rocketmq.right.listener;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.qs.rocketmq.right.dto.OrderDTO;
import com.qs.rocketmq.right.service.RightService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 定义消费者，监听消费消息
 *
 * @author qinyupeng
 * @since 2021-12-31 16:13:57
 */
@Slf4j
//@Component
//@RocketMQMessageListener(topic = "QS_ORDER_APP_TOPIC", consumerGroup = "QS_ORDER_CONSUMER_GROUP")
public class ConsumerListener implements RocketMQListener<MessageExt> {

    @Autowired
    private RightService rightService;


    /**
     * rocketMQ支持的事务消息只保证消息发送端本地事务处理和MQ消息发送一致性，而对于消费端，如果消费失败（比如：消费端本地事务处理出现异常）
     * 则消费端会不断重试（默认16次），直至消费成功。
     * 注意：判断消费端是否正常消费，取决于消费端处理是否抛出异常，可以参考源码：DefaultMessageListenerConcurrently#consumeMessage
     *
     * @author qinyupeng
     * @since 2022-01-05 16:33:30
     */
    @Override
    public void onMessage(MessageExt message) {
        log.info("消费端开始消费消息，transactionId={}，线程：{}", message.getTransactionId(), Thread.currentThread().getName());
        String keys = message.getKeys();
        if ("01".equals(keys)) {
            byte[] body = message.getBody();
            OrderDTO orderDTO = JSONUtil.toBean(StrUtil.str(body, "utf-8"), OrderDTO.class);
            log.info("开始使用券，couponId={}", orderDTO.getCouponId());
            rightService.useCoupon(orderDTO.getCouponId(), message.getTransactionId());
            log.info("券使用成功");
        }
    }
}
